package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.analyze.client.WirelessStat._
import com.lvmama.rhino.common.entity._
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._
import org.apache.spark.storage.StorageLevel

/**
  * Description:推荐转化统计清洗逻辑
  * Date:2017/3/14 18:03
  * Created by wxxuyuan
  */
object WirelessRecommendConversionRate {
  def process(sc: SparkContext, pageForward: DataFrame,date:String) = {

    //button code
    val RECOMMEND_PRODUCT_BUTTON_CODE :String = "recvst"  //新产品位的点击事件

    //page code

    //hole code

    //page entrance code

    //洗清需要的列   pageCode--->四级页面code
    val commons = sc.broadcast(Seq("logType","deviceToken", "sessionId", "channelCode", "platformCode", "pageTypeCode",
      "pageParam", "buttonCode", "pageCode","holeCode","accessCode","timestamp"))

    //进行分组的列
    val group = sc.broadcast(Seq("platform", "page", "board", "category_id","tag_id","strategy"))

    //含推荐模块的页面
    val pagesWithRec = sc.broadcast(Seq(THEME_INDEX.code,THEME_TAG_LIST.code,THEME_GUESS_ENJOY.code,ASSOCIATE_PUSH.code,DETAIL.code))
    //主题游相相关页面
    val themePages = sc.broadcast(Seq(THEME_INDEX.code,THEME_TAG_LIST.code,THEME_GUESS_ENJOY.code))

    /****** 所有推荐板块 ***
      猜你喜欢的主题游	rmdA01
      当季最受欢迎的产品推荐	rmdA02
      当季热卖门票推荐	rmdA03
      经常一起购买	rmdA04
      页面默认板块(其它产品位)	rmdA00
      */
    val allRecBoards = sc.broadcast(Seq("rmdA01","rmdA02","rmdA03","rmdA04","rmdA00","rmdA05"))


    //取出category_id、board、 tag_id，strategy
    val logWithCategoryId = pageForward.select(commons.value.map(col): _*)
    // .filter(col("userId")==="ff8080814a810d09014a85816e0101bb")
      .withColumn("category_id", when(col("pageParam").getItem("categoryId").isNull, "-1")
        .when(col("pageParam").getItem("categoryId").cast("string") === "null","-1")
        .when(col("pageParam").getItem("categoryId").cast("string") === "","-1")
        .otherwise(col("pageParam").getItem("categoryId").cast("string")))
      .withColumn("board", when(col("pageParam").getItem("board").isNull, "-1")
        .otherwise(col("pageParam").getItem("board").cast("string")))
      .withColumn("tag_id", when(col("pageParam").getItem("tagId").isNull, "-1")
        .when(col("pageParam").getItem("tagId").cast("string") === "null","-1")
        .when(col("pageParam").getItem("tagId").cast("string") === "","-1")
        .otherwise(col("pageParam").getItem("tagId").cast("string")))
      .withColumn("strategy",when(col("pageParam").getItem("strategy").isNull,"-1")
        .when(col("pageParam").getItem("strategy").cast("string")==="null","-1")
        .when(col("pageParam").getItem("strategy").cast("string")==="","-1")
        .otherwise(col("pageParam").getItem("strategy").cast("string")))

    val recommendLog = logWithCategoryId/*.filter(col("platformCode") =!= "2g53")*/
      .withColumnRenamed("platformCode","platform")
      .withColumn("product_id", when(col("pageParam").getItem("productId").isNull, "-1").otherwise(col("pageParam").getItem("productId")))
      .withColumn("pageTotalCode", concat(col("pageCode"), col("category_id"), col("product_id"), col("buttonCode"),col("board")))
      .withColumn("pageTotalCodeLag", dataLag(col("pageTotalCode")))
      .withColumn("buttonLag", dataLag(col("buttonCode")))
      .filter(col("pageTotalCodeLag").isNull.or(col("pageTotalCode") !== col("pageTotalCodeLag")))
      .filter(col("buttonCode") !== "back")
      .filter(col("buttonLag").isNull.or(col("buttonLag") !== "back"))
      .drop(col("pageTotalCode"))
      .drop(col("pageTotalCodeLag"))
      .drop(col("buttonLag"))
      .drop(col("product_id"))
      .coalesce(sc.defaultParallelism)
      .persist(StorageLevel.MEMORY_AND_DISK_SER)


    /**
      * 计算recommend_pv    推荐产品位的点击事件
      */
    val pvGroup = recommendLog.filter(col("pageTypeCode").isin(pagesWithRec.value : _*))
      .filter(col("logType").equalTo("click"))
      .filter(col("buttonCode").equalTo(RECOMMEND_PRODUCT_BUTTON_CODE))
      .filter(col("board").isNotNull && col("board") =!= "" && col("board")=!="-1")
      .withColumnRenamed("pageTypeCode","page")
      .transform(groupByColumns(group))
      .withColumnRenamed("counts", "recommend_pv")
    /**
      * 计算uv   进入到含有推荐模块的页面
      */
    val uvGroup = recommendLog.filter(col("pageTypeCode").isin(pagesWithRec.value : _*))
      .filter(col("logType") === "forward")
      .filter(col("board").isNotNull && col("board") =!= "" && col("board") =!="-1")
      .withColumnRenamed("pageTypeCode","page")
      .withColumn("deviceToken",when(col("platform")=!="2g53",col("deviceToken")).otherwise(col("sessionId")))
      .transform(groupByColumnsWithCountDistinct(group))
      .withColumnRenamed("counts", "uv")

    /**
      * 计算订单相关指标：下单用户数、订单数、订单总金额    通过支付页信息
      * 推荐途径订单条件：（accessCode有四级） 页面入口为推荐所属页面 或 页面入口为推送且bd非默认版块
      */
    val orderGroup = recommendLog.filter(col("pageTypeCode") === PENDING_PAYMENT.code)
      .withColumn("page",split(col("accessCode"),"_")(0))
      .withColumn("category_id",split(col("accessCode"),"_")(1))
      .withColumn("tag_id",split(col("accessCode"),"_")(2))
      .withColumn("board",split(col("accessCode"),"_")(3))
      .withColumn("strategy",when(split(col("accessCode"),"_")(4).isNull,"-1").when(split(col("accessCode"),"_")(4)==="null","-1").otherwise(split(col("accessCode"),"_")(4)))
      .filter(col("board").isin(allRecBoards.value : _ *)) //过滤出符合要求的数据,board非空且为有意义的推荐板块
      .filter(col("page").isin(themePages.value : _ *) ||(col("page") === ASSOCIATE_PUSH.code && col("board") =!= "rmdA00" ) || (col("page")===DETAIL.code && col("category_id")==="11" && col("board") =!= "rmdA00") )
      .withColumn("tag_id",when(col("tag_id")==="null","-1")
        .when(col("tag_id")==="","-1")
        .when(col("tag_id").isNull,"-1")
        .otherwise(col("tag_id")))//把可能为“null”的值转为“-1”
      .withColumn("category_id",when(col("category_id")==="null","-1")
      .when(col("category_id")==="","-1")
      .when(col("category_id").isNull,"-1")
      .otherwise(col("category_id")))
      .withColumn("orderId", when(col("pageParam").getItem("orderNum").isNull, "-1")
        .otherwise(col("pageParam").getItem("orderNum"))) //取出订单号
      .withColumn("orderAmount", when(col("pageParam").getItem("paymentAmount").isNull, 0)
      .otherwise(col("pageParam").getItem("paymentAmount"))) //取出订单金额
      .dropDuplicates(Seq("orderId"))
      .groupBy(group.value.map(col(_)): _*)
      .agg(countDistinct("sessionId").as("ordered_user_no"),
        count("orderId").as("order_total_no"),
        sum("orderAmount").as("order_total_amount"))


    val resultDf = pvGroup.join(uvGroup,group.value,"outer").join(orderGroup,group.value,"outer")
      .na.fill(0,Seq("recommend_pv","uv","ordered_user_no","order_total_no","order_total_amount")).withColumnRenamed("strategy","task_code")

    //获取源数据的日期
    val dateOfSouceData = date.replaceAll("/","-")

    import com.lvmama.rhino.common.utils.JDBCUtil.Conversions._
    resultDf.withColumn("date",lit(dateOfSouceData))
      .insertDF2MysqlDirect("recommend_stat")

    recommendLog.unpersist()

  }

}
